While DELETE … RETURNING is atomic, when multiple processes execute the same query simultaneously, there's a potential race condition. Using BEGIN IMMEDIATE ensures proper locking and prevents duplicate message processing.

Also, don't forget you need to handle SQLITE_BUSY error as well!

Fixed at 7e13eec and 06301ba!

The static #notifyChannels map can accumulate EventTarget instances indefinitely. Consider implementing cleanup logic when the last listener for a channel is removed.

This is a minor issue for most use cases, but worth noting for long-running applications with many different table names.

Fixed at a2851bf!

Currently, errors are caught and re-thrown without logging. Adding error logs would help with debugging:

Fixed at c1697fc!

Even with busy_timeout set, SQLITE_BUSY errors can still occur when the timeout expires. In a multi-process message queue, this is a normal condition that requires retry logic.

Consider also adding retry logic to enqueueMany() and listen() methods!

Fixed at 06301ba!

Without journal_mode=WAL, SQLite uses the default rollback journal mode which severely limits concurrent access. This will cause reader/writer conflicts in multi-process environments.

Fixed at c128708!

The regular expression #tableNameRegex allows underscores (_) in table names, but the error message only mentions "letters, digits". It would be clearer to explicitly state that underscores are also allowed.

      throw new Error(
        `Invalid table name for the message queue: ${this.#tableName}. Only letters, digits, and underscores are allowed.`,
      );

The DELETE ... RETURNING statement is wrapped in #withTransactionRetries. While this ensures atomicity and retries for the delete operation itself, the poll function's while loop continues to fetch and delete messages. If handler(message) fails, the message is re-enqueued (due to throw error at line 319), but the poll loop immediately breaks (line 327). This means that if a message consistently fails, it might block the queue from processing other messages until the next poll interval or enqueue event. Consider if the poll loop should continue attempting to process other messages even after a handler failure, or if a failing message should be moved to a dead-letter queue or have its scheduled time updated to prevent immediate re-processing.

The setTimeout in the periodic polling loop (lines 354-359) is added to timeouts but never removed if the signal aborts before the timeout fires. This could lead to a memory leak if many listen calls are made and aborted before their respective timeouts complete. The if (timeout != null) timeouts.delete(timeout); at line 360 only handles the case where the timeout does fire.

      await new Promise<unknown>((resolve) => {
        const abortListener = () => {
          if (timeout != null) clearTimeout(timeout);
          resolve();
        };
        signal?.addEventListener("abort", abortListener, { once: true });
        timeout = setTimeout(() => {
          signal?.removeEventListener("abort", abortListener);
          resolve(0);
        }, this.#pollIntervalMs);
        timeouts.add(timeout);
      });
      if (timeout != null) timeouts.delete(timeout);

The [Symbol.dispose]() method logs an error if this.#db.close() fails, but it doesn't prevent the instance from being unregistered. If close() fails, the database might still be open or in an inconsistent state, and unregistering the instance might lead to further issues or resource leaks. Consider re-throwing the error or at least making the unregistration conditional on successful closing.

The deepStrictEqual assertion here compares messages (an array) directly to ["Hello, world!"]. Given that messages are processed by multiple listeners (mq1.listen and mq2.listen), and only one listener should receive each message, the messages array might contain the message once, but the order or which listener received it isn't guaranteed. While the current test might pass due to quick processing, it's safer to assert that the message is included in the collected messages, or that the total count is correct, rather than a strict equality of the array content and order, especially in a multi-listener scenario.

Similar to the previous comment, deepStrictEqual(messages, ["Hello, world!", "Delayed message"]) might be too strict. The order of messages, especially when delays are involved and multiple listeners are active, is not strictly guaranteed to be ["Hello, world!", "Delayed message"]. It's more robust to check for the presence of both messages and the total count.

    await waitFor(() => messages.length > 1, 15_000);
    ok(messages.includes("Hello, world!"), "'Hello, world!' message should be received");
    ok(messages.includes("Delayed message"), "'Delayed message' should be received");
    deepStrictEqual(messages.length, 2, "Two messages should be received");

The assertion deepStrictEqual(new Set(messages), new Set(batchMessages)) correctly checks for the presence of all messages regardless of order. However, it doesn't explicitly verify that each message was received exactly once, which is a crucial aspect of message queue behavior (each message should be consumed by only one listener). While the testMessageQueue description mentions "Ensures messages are processed by only one listener", this specific assertion doesn't fully cover it. Consider adding a check for the total count of messages received to ensure no duplicates or missing messages.

      deepStrictEqual(new Set(messages), new Set(batchMessages));
      deepStrictEqual(messages.length, batchMessages.length, "All batch messages should be received exactly once");

Similar to the previous comment, this assertion checks for presence but not necessarily for unique consumption. Adding a length check would strengthen the test for the "processed by only one listener" guarantee.

      deepStrictEqual(new Set(messages), new Set(delayedBatchMessages));
      deepStrictEqual(messages.length, delayedBatchMessages.length, "All delayed batch messages should be received exactly once");

For the bulk enqueue stress test, deepStrictEqual(new Set(messages), expectedMessages) checks for presence. To fully ensure that each of the bulkCount messages was received exactly once and no duplicates or missing messages occurred, an additional check on the length of the messages array would be beneficial.

    deepStrictEqual(new Set(messages), expectedMessages);
    deepStrictEqual(messages.length, bulkCount, "All bulk messages should be received exactly once");